package com.sky.d_fork_jion;

import java.io.File;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.TimeUnit;

import com.google.common.collect.Lists;

/**
 * processor 	美[ˈprɑ:sesə] 中央处理器
 * 
 * 异步运行任务
 * http://ifeve.com/fork-join-4/
 * 
 * 工作窃取（work-stealing）算法是指某个线程从其他队列里窃取任务来执行
 * 
 * @Author:hjt
 */
public class FolderProcessor extends RecursiveTask<List<String>>{

	//声明这个类的序列号版本UID。这个元素是必需的，因为RecursiveTask类的父类，ForkJoinTask类实现了Serializable接口。
	private static final long serialVersionUID = 1L;
	
	//声明一个私有的、String类型的属性path。这个属性将存储任务将要处理的文件夹的全路径
	private String path;
	
	//声明一个私有的、String类型的属性extension。这个属性将存储任务将要查找的文件的扩展名。
	private String extension;
	
	//实现这个类的构造器，初始化它的属性
	public FolderProcessor(String path, String extension) {
		super();
		this.path = path;
		this.extension = extension;
	}
	
	//实现compute()方法。正如你用List<String>类型参数化RecursiveTask类，这个方法将返回这个类型的一个对象
	@Override
	protected List<String> compute() {
		//声明一个String对象的数列，用来保存存储在文件夹中的文件
		List<String> list = Lists.newArrayList();
		
		//声明一个FolderProcessor任务的数列，用来保存将要处理存储在文件夹内的子文件夹的子任务
		List<FolderProcessor> tasks = Lists.newArrayList();
		
		//获取文件夹的内容
		File file = new File(path);
		File[] contents = file.listFiles();
		
		//对于文件夹里的每个元素，如果是子文件夹，则创建一个新的FolderProcessor对象，并使用fork()方法异步地执行它
		if(contents != null){
			for(int i=0;i<contents.length;i++){
				if(contents[i].isDirectory()){
					FolderProcessor task = new FolderProcessor(contents[i].getAbsolutePath(), extension);
					task.fork();
					tasks.add(task);
				} else {
					//否则，使用checkFile()方法比较这个文件的扩展名和你想要查找的扩展名，
					//如果它们相等，在前面声明的字符串数列中存储这个文件的全路径
					if(checkFile(contents[i].getName())){
						list.add(contents[i].getAbsolutePath());
					}
				}
			}
			
			//如果FolderProcessor子任务的数列超过50个元素，写入一条信息到控制台表明这种情况
			if(tasks.size() > 50){
				System.out.printf("%s : %d tasks ran.\n", file.getAbsolutePath(), tasks.size());
			}
			//用辅助方法addResultsFromTask()，将由这个任务发起的子任务返回的结果添加到文件数列中。
			//传入参数：字符串数列和FolderProcessor子任务数列
			addResultsFromTasks(list,tasks);
		}
		//返回字符串数列
		return list;
	}

	/**
	 * 实现addResultsFromTasks()方法。
	 * 对于保存在tasks数列中的每个任务，调用join()方法，这将等待任务执行的完成，并且返回任务的结果。
	 * 使用addAll()方法将这个结果添加到字符串数列
	 */
	private void addResultsFromTasks(List<String> list, List<FolderProcessor> tasks) {
		for(FolderProcessor task : tasks){
			list.addAll(task.join());
		}
		
	}

	/**
	 * 实现checkFile()方法。这个方法将比较传入参数的文件名的结束扩展是否是你想要查找的。
	 * 如果是，这个方法返回true，否则，返回false
	 */
	private boolean checkFile(String name) {
		return name.endsWith(extension);
	}

	public static void main(String[] args) {
		//创建3个FolderProcessor任务。用不同的文件夹路径初始化每个任务
		FolderProcessor system = new FolderProcessor("C:\\Windows", "log");
		FolderProcessor apps = new FolderProcessor("C:\\Program Files","log");
		FolderProcessor documents = new FolderProcessor("C:\\Documents And Settings","log");
		
		//在池中使用execute()方法执行这3个任务
		ForkJoinPool pool = new ForkJoinPool();
		pool.execute(system);
		pool.execute(apps);
		pool.execute(documents);
		
		//将关于池每秒的状态信息写入到控制台，直到这3个任务完成它们的执行
		do {
			System.out.printf("***************\n");
			
			System.out.printf("Main: Parallelism %d\n", pool.getParallelism());
			System.out.printf("Main: Active %d\n", pool.getActiveThreadCount());
			System.out.printf("Main: Count %d\n", pool.getQueuedTaskCount());
			System.out.printf("Main: Steal Count %d\n", pool.getStealCount());
			
			System.out.printf("***************\n");
			
			try {
				//睡一秒
				TimeUnit.SECONDS.sleep(1);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		} while (!system.isDone() || !apps.isDone() || !documents.isDone());
		
		//用shutdown()方法关闭ForkJoinPool
		pool.shutdown();
		
		//将每个任务产生的结果数量写入到控制台
		List<String> results;
		results = system.join();
		System.out.printf("System: %d file found.\n", results.size());
		results = apps.join();
		System.out.printf("Apps: %d file found.\n", results.size());
		results = documents.join();
		System.out.printf("Documents: %d file found.\n", results.size());
		
	}

}

/**
 如何工作
 每个任务处理文件夹的内容
 
 如果任务找到一个文件夹，它创建另一个Task对象来处理这个文件夹，并使用fork()方法把它（Task对象）提交给池。
 这个方法提交给池的任务将被执行，如果池中有空闲的工作线程或池可以创建一个新的工作线程。
 这个方法会立即返回，所以这个任务可以继续处理文件夹的内容。
 对于每个文件，任务将它的扩展与所想要查找的（扩展）进行比较，如果它们相等，将文件名添加到结果数列。
 
 ForkJoinPool类同时允许任务的执行以异步的方式
 */
